package cn.lzd.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;


/**
 * Created by lzd on 2018/2/11.
 */
public class HdfsClient {
    private static final Logger looger = LoggerFactory.getLogger(HdfsClient.class);
    private static Configuration conf;
    private static FileSystem fileSystem;

    static {
        try {
            conf = new Configuration();
            // conf.addResource("xx.xml");
            conf.set("dfs.replication", "2");
            conf.set("dfs.blocksize", "32m");
            fileSystem = FileSystem.get(new URI("hdfs://hdp-master:9000/"), conf, "root");

        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        } finally {
            looger.error("hdfs 环境初始化错误~~");
        }
    }

    //上传
    @Test
    public void upload() {
        try {
            //fileSystem.copyFromLocalFile(new Path("d:/tms-sbj-v1.8.zip"), new Path("/sbj1.zip"));
            fileSystem.copyFromLocalFile(false, false, new Path("d:/tms-sbj-v1.8.zip"), new Path("/sbj1.zip"));
            fileSystem.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //下载
    @Test
    public void down() {
        try {
            fileSystem.copyToLocalFile(false, new Path("/sbj.zip"), new Path("d:/sbj2.zip"), true);
            fileSystem.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //创建文件夹
    @Test
    public void mkdir() {
        try {
            //创建
            fileSystem.mkdirs(new Path("/xx/xx"));
            //设置目录权限
            fileSystem.mkdirs(new Path("/xx1/xx1"), new FsPermission((short) 777));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //删除
    @Test
    public void delete() {
        try {
            //true 递归删除，可以删除文件夹
            fileSystem.delete(new Path("/xx1"), true);
            fileSystem.delete(new Path("/a.txt"), true);
            fileSystem.delete(new Path("/out.log"), true);
            fileSystem.delete(new Path("/sbj.zip"), true);
            fileSystem.delete(new Path("/sbj1.zip"), true);
            fileSystem.delete(new Path("/test"), true);
            fileSystem.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //修改名字
    @Test
    public void rename() {
        try {
            fileSystem.rename(new Path("/xx"), new Path("/xxrename"));
            fileSystem.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //查询文件,显示文件信息
    @Test
    public void listfile() {
        try {
            Configuration conf = new Configuration();
            conf.set("dfs.replication", "2");
            conf.set("dfs.blocksize", "32m");
            FileSystem fileSystem = FileSystem.get(new URI("hdfs://hdp-master:9000/"), conf, "root");
            RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fileSystem.listFiles(new Path("/"), true);
            while (locatedFileStatusRemoteIterator.hasNext()) {
                LocatedFileStatus next = locatedFileStatusRemoteIterator.next();
                System.out.println(next.getPath());
                System.out.println(Arrays.toString(next.getBlockLocations()));
                System.out.println(next.getAccessTime());
                System.out.println(next.getGroup());
                System.out.println(next.getLen());
                System.out.println(next.getModificationTime());
                System.out.println(next.getPermission());
                System.out.println(next.getOwner());
                System.out.println(next.getBlockSize());
                System.out.println(next.getReplication());
                System.out.println("--------------------------------");
            }
            fileSystem.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //查询文件夹信息
    @Test
    public void listDir() {
        try {
            /**
             * true 递归多级
             RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fileSystem.listFiles(new Path("/"), true);
             while (locatedFileStatusRemoteIterator.hasNext()){
             LocatedFileStatus fileStatus = locatedFileStatusRemoteIterator.next();
             System.out.println(fileStatus.getBlockLocations());

             }
             * */
            FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/"));

            Arrays.stream(fileStatuses).forEach((file) -> {
                System.out.println("最近访问时间：" + file.getAccessTime());
                System.out.println("文件的分块大小：" + file.getBlockSize());
                System.out.println("文件的所属者：" + file.getOwner());
                System.out.println("文件的所属组：" + file.getGroup());
                System.out.println("文件的总长度：" + file.getLen());
                System.out.println("文件的最近修改时间：" + file.getModificationTime());
                System.out.println("文件的副本数：" + file.getReplication());
                System.out.println("文件的全路径：" + file.getPath());
                System.out.println("文件的权限：" + file.getPermission());
                System.out.println(file.isDirectory() ? "d" : "f");
            });

            fileSystem.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //读取文件的部分内容
    @Test
    public void readFileblock() {
        try {
            FSDataInputStream open = fileSystem.open(new Path("/a.txt"));
//            open.seek(20); 从哪开始读。
            byte[] bt = new byte[200];
            int len = 0;
            while ((len = open.read(bt)) != -1) {
                System.out.println(new String(bt, 0, len));
            }
            open.close();
            fileSystem.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //在hdfs中创建文件并写入内容
    @Test
    public void createFileWithWrite() {
        try {
            //true 是覆盖写入，如果改成false的话，文件已存在的情况下会报错
            FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/out.log"), true);
            fsDataOutputStream.write("6666666666666666666666666666666".getBytes());
            fsDataOutputStream.flush();
            fsDataOutputStream.close();
            fileSystem.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
